package dws

import java.nio.charset.Charset
import java.util

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.hadoop.conf
import org.apache.hadoop.hbase.{Cell, HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Result, ResultScanner, Scan, Table}
import org.apache.hadoop.hbase.util.Bytes

import scala.collection.mutable

class DimAreaHBaseReader extends RichSourceFunction[(String, String)]{
  var hTable: Table = _
  val tableName: String = "dim_lagou_area"
  var connection: Connection = _
  var scan: Scan = _
  val family: String = "f1"

  override def open(parameters: Configuration): Unit = {
    val configuration: conf.Configuration = HBaseConfiguration.create()
    configuration.set("hbase.zookeeper.quorum","linux100")
    configuration.set("hbase.zookeeper.property.clientPort","2181")
    connection = ConnectionFactory.createConnection(configuration)
    hTable = connection.getTable(TableName.valueOf(tableName))
    scan = new Scan()
    scan.addFamily(family.getBytes(Charset.forName("UTF-8")))
  }

  override def run(ctx: SourceFunction.SourceContext[(String, String)]): Unit = {
    val result: ResultScanner = hTable.getScanner(scan)
    val iter: util.Iterator[Result] = result.iterator()
    while (iter.hasNext) {
      val record: Result = iter.next()
      val rowKey: Int = Bytes.toInt(record.getRow)
      val cells: util.List[Cell] = record.listCells()
      val stringBuilder = new StringBuilder
      cells.forEach(cell => {
        if ("cityId".equals(Bytes.toString(cell.getQualifierArray, cell.getQualifierOffset, cell.getQualifierLength))
          || "provinceId".equals(Bytes.toString(cell.getQualifierArray, cell.getQualifierOffset, cell.getQualifierLength))) {
          stringBuilder.append(Bytes.toInt(cell.getValueArray, cell.getValueOffset, cell.getValueLength)).append("-")
        }
        else {
          stringBuilder.append(Bytes.toString(cell.getValueArray, cell.getValueOffset, cell.getValueLength)).append("-")
        }
      })
      stringBuilder.deleteCharAt(stringBuilder.length - 1)
      ctx.collect((rowKey.toString, stringBuilder.toString()))
    }

  }

  override def cancel(): Unit = {
    try {
      if(hTable != null) {
        hTable.close()
      }
      if(connection != null) {
        connection.close()
      }
    } catch {
      case e:Exception => println(e.getMessage)
    }
  }

  def readAll(): mutable.HashMap[String, String] = {
    val configuration: conf.Configuration = HBaseConfiguration.create()
    configuration.set("hbase.zookeeper.quorum","linux100")
    configuration.set("hbase.zookeeper.property.clientPort","2181")
    connection = ConnectionFactory.createConnection(configuration)
    hTable = connection.getTable(TableName.valueOf(tableName))
    scan = new Scan()
    scan.addFamily(family.getBytes(Charset.forName("UTF-8")))
    val areaDim = new mutable.HashMap[String, String]()
    val rs: ResultScanner = hTable.getScanner(scan)
    val iterator: util.Iterator[Result] = rs.iterator()
    while (iterator.hasNext) {
      val result: Result = iterator.next()
      val rowKey: String = Bytes.toInt(result.getRow).toString
      val cells: util.List[Cell] = result.listCells()
      val stringBuilder = new StringBuilder
      cells.forEach(cell => {
        if ("cityId".equals(Bytes.toString(cell.getQualifierArray, cell.getQualifierOffset, cell.getQualifierLength))
          || "provinceId".equals(Bytes.toString(cell.getQualifierArray, cell.getQualifierOffset, cell.getQualifierLength))) {
          stringBuilder.append(Bytes.toInt(cell.getValueArray, cell.getValueOffset, cell.getValueLength)).append("-")
        }
        else {
          stringBuilder.append(Bytes.toString(cell.getValueArray, cell.getValueOffset, cell.getValueLength)).append("-")
        }
      })
      val str: String = stringBuilder.deleteCharAt(stringBuilder.length - 1).toString()
      areaDim.put(rowKey, str)
    }
    areaDim
  }
}
